﻿using System;
using System.Collections.Generic;
using System.Threading;
using Fetches.Fetch.Dto;

namespace Fetches.Fetch.PagePool.Consume
{
    /// <summary>
    /// 
    /// </summary>
    public class PagePoolConsumer
    {
        private readonly AbstractPagePool _pool;
        private readonly IProcessCoordinator _coordinator;
        private readonly IList<Action<ResponsePage>> _pageProcessors = new List<Action<ResponsePage>>();
        private int _offset = 0;
        private bool _running = true;

        public PagePoolConsumer(AbstractPagePool pool, IProcessCoordinator coordinator)
        {
            _pool = pool;
            _coordinator = coordinator;
        }

        public void AddPageProcessor(IPageProcessor processor)
        {
            _pageProcessors.Add(processor.Process);
        }

        public void AddPageProcessor(Action<ResponsePage> processor)
        {
            _pageProcessors.Add(processor);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="offset">offset小于0代表代表继续消费，其他代表从某个offset开始消费</param>
        /// <param name="processorCount"></param>
        public void StartConsume(int offset = -1, int processorCount = 1)
        {
            if (offset > -1)
            {
                _offset = offset;
            }

            _running = true;
            var endOfPool = false;
            new Thread(() =>
            {
                while (_running && !endOfPool)
                {

                    var validateOffset = _pool.ValidateOffset(_offset);
                    if (validateOffset > -1)
                    {
                        if (validateOffset > _offset)
                        {
                            _offset = validateOffset;
                        }
                        _coordinator.EnqueueWhenNoDuplicate(_offset);
                        _offset++;
                    }
                    else
                    {
                        endOfPool = _pool.IsEnd(_offset);
                        if (!endOfPool)
                        {
                            Thread.Sleep(100);
                        }
                    }
                }
            }).Start();


            for (var i = 0; i < processorCount; i++)
            {
                new Action(() =>
                {
                    //运行中pagePool或coordinator未结束就继续消费
                    while (_running && (!endOfPool || !_coordinator.IsEmpty()))
                    {
                        var pageIndex = _coordinator.Dequeue();
                        if (pageIndex > -1)
                        {
                            var page = _pool.Get(pageIndex);
                            foreach (var action in _pageProcessors)
                            {
                                action(page);
                            }
                        }

                    }
                }).Invoke();
            }
        }

        public void Suspend()
        {
            _running = false;
        }
    }
}
